package net.xuele.learn.flink.book;

import net.xuele.learn.flink.book.utils.SensorReading;
import net.xuele.learn.flink.book.utils.SensorSource;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.scala.typeutils.Types;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

/**
 * @Author patrick
 * @Date 2023/7/6 16:49
 * @Description
 */
public class TempIncreaseAlert {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.addSource(new SensorSource())
                .keyBy(value -> value.id)
                .process(new TempIncreaseAlertFunction())
                .print();
        env.execute();
    }

    /**
     * 如果某传感器的温度在1秒（处理时间）内持续增加
     * 则发出警告。
     * 这里并不是在OnTimer方法中对温度进行判断，温度判断的逻辑应该是在processElement方法中
     * <p>
     * 如果温度一直在上升，那么1s后计时器将被触发，也就是说达到了1s内温度持续上升的条件，触发告警。
     */
    private static class TempIncreaseAlertFunction extends KeyedProcessFunction<String, SensorReading, String> {

        /**
         * 用来存储上一个的温度
         */
        private ValueState<Double> lastTemp;
        /**
         * 存储当前计时器的时间
         */
        private ValueState<Long> currentTimer;

        @Override
        public void open(Configuration parameters) throws Exception {
            lastTemp = getRuntimeContext().getState(new ValueStateDescriptor<>("lastTemp", Types.DOUBLE()));
            currentTimer = getRuntimeContext().getState(new ValueStateDescriptor<>("currentTimer", Types.LONG()));
        }

        private static final long serialVersionUID = -2146929398976564695L;

        @Override
        public void processElement(SensorReading value, KeyedProcessFunction<String, SensorReading, String>.Context ctx, Collector<String> out) throws Exception {
            // 第一个过来肯定是空的（或者0.0）
            Double prevTemp = lastTemp.value();
            lastTemp.update(value.temperature);
            Long currentTimestamp = currentTimer.value();
            if (prevTemp == null || prevTemp == 0.0 || value.temperature < prevTemp) {
                // 温度下降了，删除当前计时器（可能存在计时器）
                if (currentTimestamp != null) {
                    ctx.timerService().deleteProcessingTimeTimer(currentTimestamp);
                    currentTimer.clear();
                }
            } else if (value.temperature > prevTemp && (currentTimestamp == null || currentTimestamp == 0)) {
                // 温度升高并且还未设置计时器（计时器只用设置一个即可）
                //以 当前 秒设置处理时间计时器
                long time = ctx.timerService().currentProcessingTime() + 1000;
                ctx.timerService().registerProcessingTimeTimer(time);
                // 记住当前的计时器时间，主要是为了方便删除计时器
                currentTimer.update(time);
            }
        }

        /**
         * 计时器触发了，说明在1s钟内，温度在持续上升，此时，需要收集相关的信息
         * @param timestamp
         * @param ctx
         * @param out
         * @throws Exception
         */
        @Override
        public void onTimer(long timestamp, KeyedProcessFunction<String, SensorReading, String>.OnTimerContext ctx, Collector<String> out) throws Exception {
            out.collect("alert increment " + ctx.getCurrentKey());
            currentTimer.clear();
        }
    }
}
